package com.hzg.stream.state;

import com.hzg.util.CommonUtil;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: huangzhigao
 * @Date: 2022/4/20 16:58
 */
public class BroadcastStateDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = CommonUtil.getEnv();
        DataStreamSource<Long> action = env.generateSequence(1, 10);
        DataStreamSource<Long> broad = env.generateSequence(2, 12);
        MapStateDescriptor mapState = new MapStateDescriptor("mapState", String.class, Long.class);
        BroadcastStream<Long> broadcastStream = broad.broadcast(mapState);
        Types.VOID
        action.connect(broadcastStream).process(new BroadcastProcessFunction<Long, Long, Object>() {

            @Override
            public void processElement(Long value, ReadOnlyContext ctx, Collector<Object> out) throws Exception {

            }

            @Override
            public void processBroadcastElement(Long value, Context ctx, Collector<Object> out) throws Exception {

            }
        });
    }
}
